package com.atguigu.bigdata.chapter05.source;

import com.atguigu.bigdata.bean.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.Socket;
import java.nio.charset.StandardCharsets;

/**
 * @Author lzc
 * @Date 2022/9/2 10:01
 */
public class Flink04_Source_Custom {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        env.addSource(new MySocketSource()).print();
        
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    // 自定义一个可以从socket读取数据的source
    public static class MySocketSource implements SourceFunction<WaterSensor> {
        
        boolean isStop = false;
        
        // 在这个方格内部读取数据源的的数据, 读完解析处理之后, 写入到后序的流中
        @Override
        public void run(SourceContext<WaterSensor> ctx) throws Exception {
            Socket socket = new Socket("hadoop162", 9999);
            // 从网络中获取自己输入流
            InputStream is = socket.getInputStream();
            BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8));
            String line = reader.readLine();
            
            while (!isStop && line != null) {
                // 对读到的一行数据做处理
                String[] data = line.split(",");
                // 把数据放入到流中
                ctx.collect(new WaterSensor(data[0],
                                            Long.valueOf(data[1]),
                                            Integer.valueOf(data[2])
                            )
                );
                
                line = reader.readLine();
            }
            
            
        }
        
        // 取消: 关闭source
        // 这个方法不会自动调用, 如果有需要的时候, 外部可以通过调用这个方法实现关闭source
        @Override
        public void cancel() {
            isStop = true;
        }
    }
}
